package com.aotain.nyx.threateva;

import java.sql.Connection;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map.Entry;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.apache.log4j.Logger;

import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.common.CommonDB;
import com.aotain.common.DbPoolMultiple;
import com.aotain.common.mongo.DataColumn;
import com.aotain.nyx.ipstat.IPStatTuple;
import com.aotain.nyx.statis.AbnStatisTuple;

public class ThreatenEvaluateImport implements FlatMapFunction<Tuple2<String,AbnStatisTuple>	, Object> {

	/** 
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	 */ 
	private static final long serialVersionUID = 6706009868030895225L;
	
	private String dbJson;
	public ThreatenEvaluateImport(String dbJson){
		this.dbJson = dbJson;
	}

	@Override
	public void flatMap(Tuple2<String, AbnStatisTuple> value,
			Collector<Object> out) throws Exception {
		// TODO Auto-generated method stub
		java.sql.Statement stmt = null;
		Connection conn = DbPoolMultiple.getInstance().getConnJson(dbJson);
		String insert = "insert into METIS_THREATEN_EVALUATE (REPORTTIME,IP,EVALUATE) values ('%s','%s','%s')";
		SimpleDateFormat sdfh_ = new SimpleDateFormat("yyyy-MM-dd HH:00:00");
		Date date = new Date();
		String dateStrH_ = sdfh_.format(date);
		String sip = "";
		Integer eval = 0;
		
		try {
			
			stmt = conn.createStatement();
			conn.setAutoCommit(false);
				
			sip = value.f1.getSourceIP();
			eval = value.f1.getEvaluate();
			String sql = String.format(insert, new Object[]{dateStrH_,sip,eval});
			stmt.execute(sql);
				
			conn.commit();
			
		} catch (SQLException e) {
			Logger.getLogger(getClass()).error("insert ip list error",e);
		} finally {
			CommonDB.closeDBConnection(conn, stmt, null);
		}
		
		
	}

}
